package com.ibm.spatiotemp.utils;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.Compressor;
import org.apache.hadoop.io.compress.Decompressor;
import org.apache.hadoop.io.compress.DecompressorStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.log4j.Logger;

import com.ibm.spatiotemp.server.config.LocalConfig;

public class StringEncoder {
	
	public static Logger logger = Logger.getLogger(StringEncoder.class);
	
	private static Base64 base64Decoder = new Base64();
	
	public static String codecClassname_bzip2 = "org.apache.hadoop.io.compress.bzip2Codec"; 
	public static String codecClassname2_snappy = "org.apache.hadoop.io.compress.SnappyCodec";
	public static String codecClassname_gzip = "org.apache.hadoop.io.compress.GzipCodec"; 
	public static String codecClassname_lz4 = "org.apache.hadoop.io.compress.lz4Codec"; 
	
	public static String compress(String method, String inputstring) throws IOException, ClassNotFoundException { 
		logger.info("inputstring length:" + inputstring.length());
		byte[] orgbytes = inputstring.getBytes(LocalConfig.String_Class);
		logger.info("orgbyte len:" + orgbytes.length);
		byte[] encodedbytes = base64Decoder.encodeBase64(orgbytes);
		logger.info("base64 encodedbytes len:" + encodedbytes.length);
		
		 Class<?> codecClass = Class.forName(method);  
		 Configuration conf = new Configuration();  
		 CompressionCodec codec = (CompressionCodec)  
		        ReflectionUtils.newInstance(codecClass, null);  
		 
		 Compressor compressor = codec.createCompressor();
		 ByteArrayOutputStream compressed_byte_stream = new ByteArrayOutputStream();
		 CompressionOutputStream cout = codec.createOutputStream(compressed_byte_stream, compressor);
		 cout.write(encodedbytes);
		 cout.close();
		 
		 byte[] compressed = compressed_byte_stream.toByteArray();
		 logger.info("compressed len:" + compressed.length);
		 
		 byte[] b64encoded_bytes = base64Decoder.encode(compressed);
		 String result = new String(b64encoded_bytes);
		 logger.info("compressed string length:" + result.length());
		 return result;
		}  
	
	
	public static String decompress(String method, String compressedstr) throws IOException, ClassNotFoundException {  
	  Configuration conf = new Configuration();  
//	  CompressionCodecFactory factory = new CompressionCodecFactory(conf);  
	 
	  Class<?> codecClass = Class.forName(method); 
	  CompressionCodec codec = (CompressionCodec)  
		        ReflectionUtils.newInstance(codecClass, null);  
	  
	  	 byte[] in_base64_bytes = base64Decoder.decode(compressedstr.getBytes());
		 Decompressor dcompressor = codec.createDecompressor();
		 ByteArrayInputStream bin = new ByteArrayInputStream(in_base64_bytes);

//		 CompressionInputStream cin = codec.createInputStream(bin, dcompressor);
		 CompressionInputStream cin  = new DecompressorStream(bin, dcompressor,4*1024);
		 
		 ByteArrayOutputStream bout = new ByteArrayOutputStream();
		 IOUtils.copy(cin, bout, 4096);
		 
		 byte[] bout_bytes = bout.toByteArray();
		 byte[] raw_bytes = base64Decoder.decodeBase64(bout_bytes);
		 
		 String result = new String(raw_bytes,LocalConfig.String_Class);
		 return result;
	}
	
	
	public static void main(String[] args)
	{
		String teststring = "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		teststring += "asdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjkasdfghjklasdfghjk";
		
		try {
			String compressstr = compress(LocalConfig.codecClassname, teststring);
			
			logger.info(compressstr);
			
			String dcmpedstr = decompress(LocalConfig.codecClassname, compressstr);
			
			logger.info(dcmpedstr);
		} catch (ClassNotFoundException | IOException e) {
			e.printStackTrace();
		}
		
	}
	
	
}
